package cn.monkey.transport.mq.rabbitmq;

import cn.money.transport.mq.MqOutboundConfig;
import com.rabbitmq.client.Channel;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Supplier;

public class SimpleRabbitMqOutbound implements RabbitMqOutbound {

    private final RabbitMqOutboundConfig outboundConfig;
    protected final BlockingQueue<Object> cache;
    protected final Collection<Channel> channels;
    protected final Supplier<Mono<Void>> closeAction;

    public SimpleRabbitMqOutbound(Collection<Channel> channels,
                                  RabbitMqOutboundConfig outboundConfig,
                                  Supplier<Mono<Void>> closeAction) {
        this.outboundConfig = outboundConfig;
        this.channels = channels;
        this.closeAction = closeAction;
        int cacheSize = outboundConfig.getCacheSize();
        this.cache = cacheSize > 0 ? new ArrayBlockingQueue<>(cacheSize) : null;
    }

    @Override
    public MqOutboundConfig outboundConfig() {
        return this.outboundConfig;
    }

    @Override
    public Mono<Void> close() {
        return this.closeAction.get();
    }


    @Override
    public Mono<Void> write(Object o) {
        if (this.cache == null) {
            return writeAndFlush(o);
        }
        try {
            this.cache.add(o);
            return Mono.empty();
        } catch (Exception e) {
            return Mono.error(e);
        }
    }

    @Override
    public Mono<Void> writeAndFlush(Object o) {
        List<CompletableFuture<Void>> list = Arrays.stream(this.outboundConfig.getTopics())
                .map(topic -> CompletableFuture.runAsync(() -> {
                    try {
                        for (Channel channel : this.channels) {
                            channel.basicPublish(topic, topic, null, this.outboundConfig.getEncoder().encode(o));
                        }
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }))
                .toList();
        return Mono.fromCompletionStage(CompletableFuture.allOf((CompletableFuture<?>) list));
    }

    @Override
    public Mono<Void> flush() {
        if (this.cache == null || this.cache.isEmpty()) {
            return Mono.empty();
        }
        List<Object> objects = new ArrayList<>(this.cache.size());
        this.cache.drainTo(objects);
        List<Mono<Void>> list = objects.stream()
                .map(this::writeAndFlush)
                .toList();
        return Flux.merge(list).then();
    }
}
